prediction_market_agent/db/blockchain_transaction_fetcher.py (1)

Line range hint 44-70: Improve performance by avoiding row-wise iteration over the DataFrame

Using df.iter_rows(named=True) for converting DataFrame rows to BlockchainMessage objects can be inefficient for large datasets. Polars supports vectorized operations, which are more performant.

Consider using df.apply or vectorized methods to convert the DataFrame to a list of BlockchainMessage objects without explicit iteration:

return [
-    self.blockchain_message_from_dune_df_row(consumer_address, x)
-    for x in df.iter_rows(named=True)
+    self.blockchain_message_from_dune_df_row(consumer_address, row)
+    for row in df.to_dicts()
]

This approach leverages Polars' efficient data handling capabilities.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

prediction_market_agent/db/blockchain_transaction_fetcher.py (1)

Line range hint 44-72: Protect against SQL injection and optimize query performance

The SQL query construction using string formatting is vulnerable to SQL injection. Additionally, fetching all columns (*) may impact performance unnecessarily.

1. Use parameterized queries to prevent SQL injection:

-query = f'select * from gnosis.transactions where "to" = {Web3.to_checksum_address(consumer_address)} AND block_number >= {min_block_number} and value >= {xdai_to_wei(MicrochainAgentKeys().RECEIVER_MINIMUM_AMOUNT)} order by block_time asc'
+query = '''
+    SELECT hash, value, block_number, "from", data, block_time 
+    FROM gnosis.transactions 
+    WHERE "to" = :consumer_address 
+    AND block_number >= :min_block_number 
+    AND value >= :min_value 
+    ORDER BY block_time ASC
+'''
+params = {
+    "consumer_address": Web3.to_checksum_address(consumer_address),
+    "min_block_number": min_block_number,
+    "min_value": xdai_to_wei(MicrochainAgentKeys().RECEIVER_MINIMUM_AMOUNT)
+}
+df = spice.query(query, params, api_key=keys.dune_api_key.get_secret_value())

2. Consider adding a LIMIT clause to prevent memory issues with large result sets:

# Add to query string:
LIMIT 1000  # Adjust based on your needs

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro

 def blockchain_message_from_dune_df_row(
     self, consumer_address: ChecksumAddress, x: dict[str, Any]
 ) -> BlockchainMessage:
+    try:
+        # Validate required fields exist and have correct types
+        required_fields = {"hash", "value", "block_number", "from", "data"}
+        if missing := required_fields - x.keys():
+            raise ValueError(f"Missing required fields: {missing}")
+            
         return BlockchainMessage(
             consumer_address=consumer_address,
-            transaction_hash=x["hash"],
-            value_wei=str(x["value"]),
-            block=str(x["block_number"]),
-            sender_address=x["from"],
-            data_field=self.unzip_message_else_do_nothing(x["data"]),
+            transaction_hash=x.get("hash", ""),
+            value_wei=str(x.get("value", 0)),
+            block=str(x.get("block_number", 0)),
+            sender_address=x.get("from", ""),
+            data_field=self.unzip_message_else_do_nothing(x.get("data", "")),
         )
+    except Exception as e:
+        raise ValueError(f"Failed to create BlockchainMessage from row: {e}") from e

 def fetch_one_unprocessed_blockchain_message_and_store_as_processed(
     self, consumer_address: ChecksumAddress
 ) -> BlockchainMessage | None:
     """
     Method for fetching oldest unprocessed transaction sent to the consumer address.
     After being fetched, it is stored in the DB as processed.
     """
-    transactions = self.fetch_unseen_transactions(consumer_address=consumer_address)
+    transactions = self.fetch_unseen_transactions(consumer_address=consumer_address, limit=1)
     if not transactions:
         return None

     blockchain_message = transactions[0]

-    # mark unseen transaction as processed in DB
-    self.blockchain_table_handler.save_multiple([blockchain_message])
+    try:
+        # mark unseen transaction as processed in DB
+        self.blockchain_table_handler.save_multiple([blockchain_message])
+    except Exception as e:
+        # Log the error but don't raise to prevent reprocessing the same transaction
+        logger.error(f"Failed to mark transaction as processed: {e}")
+
     return blockchain_message

prediction_market_agent/db/blockchain_transaction_fetcher.py (1)

Line range hint 44-70: Prevent SQL injection by using parameterized queries

The current implementation uses string interpolation for SQL query parameters, which could be vulnerable to SQL injection attacks. While the risk might be lower since the inputs are processed through Web3.to_checksum_address(), it's still best practice to use parameterized queries.

Consider using Dune API's parameter binding mechanism if available, or at minimum, implement proper input sanitization:

-query = f'select * from gnosis.transactions where "to" = {Web3.to_checksum_address(consumer_address)} AND block_number >= {min_block_number} and value >= {xdai_to_wei(MicrochainAgentKeys().RECEIVER_MINIMUM_AMOUNT)} order by block_time asc'
+# Example using parameters (adjust syntax based on spice/Dune API capabilities)
+query = '''
+    select * from gnosis.transactions 
+    where "to" = :address 
+    AND block_number >= :min_block 
+    and value >= :min_value 
+    order by block_time asc
+'''
+params = {
+    "address": Web3.to_checksum_address(consumer_address),
+    "min_block": min_block_number,
+    "min_value": xdai_to_wei(MicrochainAgentKeys().RECEIVER_MINIMUM_AMOUNT)
+}
+df = spice.query(query, params, api_key=keys.dune_api_key.get_secret_value())

Additionally, consider adding error handling for the Dune API call and documentation for the query's assumptions about block ordering.

prediction_market_agent/db/blockchain_transaction_fetcher.py (3)

32-42: ⚠️ Potential issue

Handle potential KeyError exceptions when accessing dictionary keys

Direct dictionary access without validation could cause crashes when processing incomplete data from Dune API.

Previous review comment still applies - consider using dictionary.get() with default values to handle missing keys safely.

---

75-76: 🛠️ Refactor suggestion

Optimize transaction counting to avoid fetching full records

Previous review comment about performance optimization still applies - fetching all transaction data just to get the count is inefficient. Consider implementing the suggested COUNT query approach.

---

85-94: 🛠️ Refactor suggestion

Optimize single message fetching to avoid retrieving all transactions

Previous review comment about optimizing single transaction fetching still applies. Consider implementing the suggested LIMIT 1 approach to avoid unnecessary data retrieval.

Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro